import asyncio

import talib
import pandas as pd
from collections import deque
from typing import Dict, Deque, Callable

from longport.openapi import PushQuote
from numpy import float64

from config import config



class MarketDataProcessor:
    def __init__(self, symbols: list):
        self.loop = asyncio.get_event_loop()
        self.symbols = symbols
        self.tick_buffers: Dict[str, Deque] = {sym: deque(maxlen=1000) for sym in symbols}
        self.bar_data: Dict[str, pd.DataFrame] = {sym: pd.DataFrame() for sym in symbols}
        self.indicators: Dict[str, dict] = {sym: {} for sym in symbols}

        self._indicators_updated_callbacks: Dict[str, list] = {}

        # 初始化指标参数
        self.ema_period = 12
        self.rsi_period = 14
        self.adx_period = 14
        self.optional_period = 35
        self.min_data_length = max(self.ema_period, self.rsi_period, self.adx_period * 2, self.optional_period)

        # 计算中间状态
        self.current_bar_start_volume: Dict = {sym: None for sym in symbols}

        # self.indicators: Dict[str, dict] = {
        #     sym: {
        #         'ema': [],
        #         'rsi': [],
        #         'macd_hist': [],
        #         'adx': [],
        #         'close_history': []  # 存储原始收盘价序列
        #     } for sym in symbols
        # }

    def on_market_data(self, symbol: str, quota_data: PushQuote):
        # print(symbol, quota_data)
        # 异步更新数据
        asyncio.run_coroutine_threadsafe(self.process_tick(symbol, quota_data), self.loop)






    async def process_tick(self, symbol: str, tick: PushQuote):
        # print(symbol, tick)
        """处理实时tick数据"""
        self.tick_buffers[symbol].append(tick)
        # # #
        # 当累计足够tick时生成bar
        if len(self.tick_buffers[symbol]) % config.bar_size == 0:
            await self._create_bar(symbol)

    async def _create_bar(self, symbol: str):

        """生成OHLC Bar"""
        if len(self.tick_buffers[symbol]) < config.bar_size:
            return

        ticks = list(self.tick_buffers[symbol])[-config.bar_size:]
        # print(f"start create_bar for {symbol} with ticks {ticks}")
        bar = {
            'timestamp': ticks[-1].timestamp,
            'open': float64(ticks[0].last_done),
            'high': float64(max(t.last_done for t in ticks)),
            'low': float64(min(t.last_done for t in ticks)),
            'close': float64(ticks[-1].last_done),
            'volume': float64(ticks[-1].volume - (self.current_bar_start_volume[symbol] or ticks[0].volume))
        }

        # 更新DataFrame
        new_row = pd.DataFrame([bar])
        self.bar_data[symbol] = pd.concat(
            [self.bar_data[symbol], new_row],
            ignore_index=True
        ).iloc[-100:]  # 保留最近100个bar

        # 更新当前bar的起始量
        self.current_bar_start_volume[symbol] = ticks[-1].volume


        # 计算技术指标
        await self._calculate_indicators(symbol)

    async def _calculate_indicators(self, symbol: str):
        # print(f"start calculate_indicators for {symbol}")
        """计算技术指标"""
        df = self.bar_data[symbol]
        # print(f"len of df is {len(df)}")
        if len(df) < self.min_data_length:
            # print(f"Not enough data for {symbol}, {len(df)} < {self.min_data_length}")
            return
        # print(symbol)
        # print(self.bar_data[symbol])
        # import pdb;pdb.set_trace()
        closes = df['close'].values
        highs = df['high'].values
        lows = df['low'].values
        volumes = df['volume'].values

        # 计算MACD
        macd, macd_signal, macd_hist = talib.MACD(
            closes,
            fastperiod=12,
            slowperiod=26,
            signalperiod=9
        )
        ema_fast = talib.EMA(closes, timeperiod=12)
        # print(f"{symbol}, ema_fast {ema_fast}")
        ema_slow= talib.EMA(closes, timeperiod=26)
        adx = talib.ADX(highs, lows, closes, timeperiod=14)
        rsi = talib.RSI(closes, timeperiod=14)
        slowk, slowd = talib.STOCH(highs, lows, closes, fastk_period=9, slowk_period=3, slowd_period=3)
        atr = talib.ATR(highs, lows, closes, timeperiod=14)
        upper, middle, lower = talib.BBANDS(closes, timeperiod=20, nbdevup=2, nbdevdn=2)
        obv = talib.OBV(closes, volumes)
        obv_ma20 = talib.MA(obv, timeperiod=20)

        tp = talib.TYPPRICE(highs, lows, closes)
        cumulative_tp = (tp * volumes).cumsum()
        vwap = cumulative_tp / volumes.cumsum()
        # 更新指标
        self.indicators[symbol] = {
            'ema_fast': ema_fast[-1],
            'ema_slow': ema_slow[-1],
            'rsi':rsi[-1],
            'macd': macd[-1],
            'macd_signal': macd_signal[-1],
            'macd_hist': macd_hist[-1],
            'adx': adx[-1],
            'atr': atr[-1],
            'bb_width': ((upper - lower) / middle)[-1],
            'obv': obv[-1],
            'obv_ma20': obv_ma20[-1],
            'vwap': vwap[-1],
            'close': closes[-1],
        }
        # self.indicators[symbol]['ema_fast'] = ema_fast
        # self.indicators[symbol]['ema_slow'] = ema_slow
        # self.indicators[symbol]['rsi'] = rsi
        # self.indicators[symbol]['macd'] = macd
        # self.indicators[symbol]['macd_signal'] = macd_signal
        # self.indicators[symbol]['macd_hist'] = macd_hist
        # # print(f"type of macd_hist is {type(macd_hist)}")
        # self.indicators[symbol]['adx'] = adx
        # self.indicators[symbol]['atr'] = atr
        # self.indicators[symbol]['bb_width'] = ((upper - lower) / middle)
        # self.indicators[symbol]['obv'] = obv
        # self.indicators[symbol]['obv_ma20'] = obv_ma20
        # self.indicators[symbol]['vwap'] = vwap

        # print(symbol)
        # print(self.indicators[symbol])
        # 触发回调
        if symbol in self._indicators_updated_callbacks:
            for callback in self._indicators_updated_callbacks[symbol]:
                # print(f"{symbol} updated")
                # print(callback)
                await callback(self.indicators[symbol], symbol)

    # 在MarketDataProcessor中添加
    @property
    def last_prices(self) -> Dict[str, float]:
        """最新价格缓存"""
        return {sym: ticks[-1].last_done for sym, ticks in self.tick_buffers.items()}

    def register_indicators_updated_callback(self, symbol: str, callback: Callable):
        """注册指标更新回调"""
        if symbol not in self._indicators_updated_callbacks:
            self._indicators_updated_callbacks[symbol] = []
        self._indicators_updated_callbacks[symbol].append(callback)



